package com.niit;

import com.Entity.Person;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class ProducerDemo3 {

        public static void main(String[] args) {
                // 1. 准备生产者的配置对象
                Properties props = new Properties();
                //kafka 集群，broker-list
                props.put("bootstrap.servers", "niit01:9092");
                props.put(ProducerConfig.ACKS_CONFIG, "all"); //acks=1
                //重试次数
                props.put("retries", 1);

                //批次大小
                props.put("batch.size", 16384);
                //等待时间
                props.put("linger.ms", 1);
                //RecordAccumulator 缓冲区大小
                props.put("buffer.memory", 33554432);
                props.put("key.serializer",
                        "org.apache.kafka.common.serialization.StringSerializer");
                props.put("value.serializer",
                        "com.niit.PersonSerdes");
                // 2. 初始化生产者对象
                Producer<String, Person> producer = new KafkaProducer<>(props);
                // 3. 发布数据
                for (int i = 0; i < 50; i++) {
                    Person person = new Person(i, "sam" + i, i, 5000 + i);
                    producer.send(new ProducerRecord<String, Person>("test1", 0, Integer.toString(i), person));

                }
                // 4. 释放资源
                producer.flush();
                producer.close();
            }


        }
